package com.leyantech.ecp.agenx.dispatcher.client.di;

import com.leyantech.utility.constructs.Constructs;

import com.ctrip.framework.apollo.Config;
import com.ctrip.framework.apollo.ConfigService;
import com.google.common.collect.Maps;
import com.google.inject.AbstractModule;
import com.google.inject.TypeLiteral;
import com.google.inject.name.Names;
import io.opentracing.contrib.kafka.TracingKafkaProducer;
import io.opentracing.util.GlobalTracer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Map;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-12-05.
 */
public class AgenxDispatchClientModule extends AbstractModule {

  @Override
  protected void configure() {

    bind(new TypeLiteral<Producer<String, String>>() {
    })
        .annotatedWith(Names.named("dispatch-client"))
        .toInstance(provideProducer());

  }

  protected Producer<String, String> provideProducer() {
    final Config apolloConfig = ConfigService.getConfig("1.ecp-agenx-pdd");
    final String brokerServer = apolloConfig.getProperty("broker.list", "");

    final Map<String, Object> configs = Maps.newHashMap();

    configs.put("acks", "all");
    configs.put("retries", 3);
    configs.put("request.timeout.ms", 10000);
    configs.put("bootstrap.servers", brokerServer);
    configs.put("client.id", String.join(".", Constructs.LAIN_APPNAME, Constructs.HOSTNAME));
    configs.put("compression.type", "snappy");
    configs.put("key.serializer", StringSerializer.class);
    configs.put("value.serializer", StringSerializer.class);

    return new TracingKafkaProducer<>(new KafkaProducer<>(configs), GlobalTracer.get());
  }

}
